# -*- coding: utf-8 -*-
# @Time    : 2025/10/14
# @Author  : 楚小涛
# @File    : eventbus_memory.py

"""
进程内事件总线（内存实现，线程安全）
- 使用 queue.Queue（线程安全）承载每个订阅通道，适配“后台执行器线程”→“SSE 同步线程”跨线程传递
- 订阅/退订：同步（threading.RLock）
- 发布：非阻塞 put_nowait，满队列按策略处理（丢最旧或丢最新）
- 提供 stats/snapshot 便于调试
"""

from __future__ import annotations

import queue
import threading
from collections import defaultdict
from typing import Any, Dict, List, Sequence, Tuple

from execution.interfaces import SubscribableEventBus, EventPayload
from domain.enums import EventType


class MemoryEventBus(SubscribableEventBus):
    """
    进程内事件总线（支持订阅），线程安全。
    """

    # 队列溢出策略：'drop_oldest' 或 'drop_new'
    QUEUE_OVERFLOW_POLICY = "drop_oldest"

    def __init__(self, *, max_queue_size: int = 1000) -> None:
        # exec_id -> List[queue.Queue]
        self._subs: Dict[str, List[queue.Queue]] = defaultdict(list)
        self._lock = threading.RLock()
        self._closed = False
        self._maxsize = max_queue_size

        # 统计信息
        self._events_published = 0
        self._events_dropped = 0

    # ----------------------------
    # 订阅管理（同步）
    # ----------------------------

    def create_queue(self) -> queue.Queue:
        """创建一个带上限的线程安全队列供订阅端使用"""
        return queue.Queue(maxsize=self._maxsize)

    def subscribe(self, exec_id: str, queue_like: Any) -> None:
        """注册订阅者（同步，兼容 Django 同步视图）"""
        if self._closed:
            raise RuntimeError("MemoryEventBus 已关闭。")
        if not isinstance(queue_like, queue.Queue):
            raise TypeError("queue_like 必须是 queue.Queue（线程安全队列）。")
        with self._lock:
            self._subs[exec_id].append(queue_like)
        print(f"[BUS] subscribe exec={exec_id} bus_id={id(self)} subs={len(self._subs[exec_id])}")

    def unsubscribe(self, exec_id: str, queue_like: Any) -> None:
        """取消订阅（同步）"""
        with self._lock:
            lst = self._subs.get(exec_id, [])
            if queue_like in lst:
                lst.remove(queue_like)
            if not lst and exec_id in self._subs:
                self._subs.pop(exec_id, None)
        print(f"[BUS] unsubscribe exec={exec_id} bus_id={id(self)}")

    # ----------------------------
    # 发布事件（快速、非阻塞）
    # ----------------------------

    def publish(self, exec_id: str, event: EventType, payload: EventPayload) -> None:
        """单条事件广播到该 exec 的所有订阅队列"""
        if self._closed:
            return

        record = {"event": event.value, "payload": payload}
        with self._lock:
            queues = list(self._subs.get(exec_id, []))

        # 调试打印
        print(f"[BUS] publish exec={exec_id} bus_id={id(self)} subs={len(queues)}")

        for q in queues:
            self._safe_put_nowait(q, record)
        self._events_published += 1

    def publish_many(self, records: Sequence[Tuple[str, EventType, EventPayload]]) -> None:
        """批量发布（按 exec_id 广播）"""
        if self._closed or not records:
            return

        # 先抓订阅快照，减少锁粒度
        with self._lock:
            subs_snapshot: Dict[str, List[queue.Queue]] = {}
            for (eid, _, _) in records:
                if eid not in subs_snapshot:
                    subs_snapshot[eid] = list(self._subs.get(eid, []))

        for exec_id, event, payload in records:
            item = {"event": event.value, "payload": payload}
            for q in subs_snapshot.get(exec_id, []):
                self._safe_put_nowait(q, item)
            self._events_published += 1

    def _safe_put_nowait(self, q: queue.Queue, item: dict) -> None:
        """
        非阻塞入队；满了按策略处理。
        - drop_oldest：丢弃最旧的一条，再尝试放入最新
        - drop_new：直接丢弃新来的
        """
        try:
            q.put_nowait(item)
            return
        except queue.Full:
            pass

        if self.QUEUE_OVERFLOW_POLICY == "drop_oldest":
            # 丢掉一个最旧的
            try:
                _ = q.get_nowait()
            except queue.Empty:
                # 极端情况下：刚满→被其他线程消费→再尝试放入
                pass
            else:
                # 计数丢弃
                self._events_dropped += 1
            # 再次尝试放新项
            try:
                q.put_nowait(item)
            except queue.Full:
                # 仍然放不进去，算丢弃新项
                self._events_dropped += 1
        else:
            # drop_new：直接丢新项
            self._events_dropped += 1

    # ----------------------------
    # 资源管理 / 观测
    # ----------------------------

    def close(self) -> None:
        with self._lock:
            self._closed = True
            self._subs.clear()

    def stats(self) -> dict:
        with self._lock:
            subs = sum(len(v) for v in self._subs.values())
        return {
            "events_published": self._events_published,
            "events_dropped": self._events_dropped,
            "subscribers": subs,
            "closed": self._closed,
        }

    def snapshot(self) -> dict:
        """调试快照：每个 exec_id 的订阅数量及队列长度"""
        with self._lock:
            channels = []
            for eid, qs in self._subs.items():
                channels.append({
                    "exec_id": eid,
                    "subscribers": len(qs),
                    "queues": [{"index": i, "qsize": q.qsize()} for i, q in enumerate(qs)]
                })
        return {
            "bus_id": id(self),
            "events_published": self._events_published,
            "events_dropped": self._events_dropped,
            "channels": channels,
        }
